<?php
namespace xlx;

use Workerman\Events\EventInterface;
 
class DispatchClient
{
    //public: 
    public static function connect($ip='127.0.0.1', $port=8109)
    {
        self::$server_name = "$ip:$port";
        $event = Util::getEventLoop();
        $conn =  self::getTcpServer();
        self::$client_name = $conn->getRemoteIp().':'.$conn->getRemotePort();
        return $conn;
    }
    public static function getServerName()
    {
        if (empty(self::$server_name)) {
            throw new Exception("Error Processing Request", 1);
        }
        return self::$server_name;
    }
    public static function getClientName()
    {
        if (empty(self::$client_name)) {
            throw new \Exception("Error Processing Request", 1);
        }
        return self::$client_name;
    }
 
    //protected: 
    protected static $server_name;
    protected static $client_name;
    protected static $message=[];
    protected static $tcp_peers=[];
    protected static $tcp_listen;
    protected static $lasttime;
    protected static function getTcpReceiver()
    {
        $event = Util::getEventLoop();
        if (empty(self::$tcp_listen)) {
            self::$tcp_listen = @stream_socket_server('tcp://'.self::getClientName());
            $event->add(self::$tcp_listen, EventInterface::EV_READ, function ($socket) {
                $socket = \stream_socket_accept($socket, 0, $addr);
                $remote_addr = stream_socket_get_name($socket, true);
                self::createPeer($socket, $remote_addr);
            });
        }
        return $event;
    }
    protected static function getTcpServer()
    {
        if (empty(self::$server_name)) {
            throw new \Exception("Error Processing Request", 1);
        }
        return self::getTcpPeer(self::$server_name);
    }
    protected static function getTcpPeer($addr)
    {
        if (empty(self::$tcp_peers[$addr])) {
            $s = @stream_socket_client("tcp://{$addr}", $en, $es);
            if (empty($s)) {
                throw new \Exception("Error Processing Request", 1);
            }
            self::createPeer($s, $addr);
        }
        return self::$tcp_peers[$addr];
    }
    protected static function createPeer($socket, $addr)
    {
        if (isset(self::$tcp_peers[$addr])) {
            throw new \Exception("Error Processing Request", 1);
        }
        $local_addr = stream_socket_get_name($socket, false);
        self::$tcp_peers[$addr] = new \Workerman\Connection\TcpConnection( $socket, $local_addr );
        self::$tcp_peers[$addr]->protocol = \Protocols\Json::class;
        self::$tcp_peers[$addr]->onClose = function ($conn) use ($addr) {
            unset(self::$tcp_peers[$addr]);
        };
        self::$tcp_peers[$addr]->onMessage = function ($conn, $data) {
            //callback receive&message 
            switch ($data[0]) {
                case 'spawn_resolve'://0:TOKEN 1:PID 2:VLAUE
                case 'spawn_reject'://0:TOKEN 1:PID 2:VLAUE
                case 'send'://0:TOKEN 1:PID 2:VLAUE
                    if (isset(self::$message[$data[1]])) {
                        $defer = self::$message[$data[1]];
                        if ($data[0]=='spawn_reject' && !$defer->trap_exit) {
                            $defer->reject(new \Error($data[2]));
                        }
                        if ($ret = \call_user_func($defer->selectfn, $data[2])) {
                            $defer->resolve($ret);
                        }
                    }
                    break;
                case 'esmdo_resolve'://0:TOKEN 1:PID 2:STATE 3:VALUE
                case 'esmdo_reject': //0:TOKEN 1:PID 2:STATE 3:VALUE
                case 'esm_send'://0:TOKEN 1:PID 2:STATE 3:VALUE
                    if (isset(self::$message[$data[1]][$data[2]])) {
                        $fn = (self::$message[$data[1]][$data[2]]);
                        $fn($data[1], $data[3]);
                    }
                    break;
                default:
                    throw new \Exception(print_r($data, true), 1);
            }
        };
    }
    public static function startWorker($opt)
    {
        self::$server_name = "{$opt['ip_addr']}:{$opt['ip_port']}";
        $event = Util::getEventLoop();
        $server = self::getTcpServer();
        self::$client_name = $server->getRemoteIp().':'.$server->getRemotePort();
        $server->onClose = function () {
            //echo "WORKER EXIT\n";
            exit;
        };
        $server->onMessage = function ($conn, $data) {
            self::$lasttime=time();
            //echo "\nC:".substr(json_encode($data),0,128) ;
            switch ($data[0]) {
                case 'spawn'://0:TOKEN 1:FN 2:ARGS 3:OPT
                    try {
                        $result = call_user_func_array( Util::decodeClosure($data[1]), $data[2] );
                    } catch (\Throwable $e) {
                    }
                    self::getTcpServer()->send(['done']);
                    if (!empty($data[3]['link'])) {
                        $pid = $data[3]['link'];
                        if (preg_match('/([\w\.:]+)@(\d+)/', $pid, $arr)) {
                            list($_,$addr,/*uniquid*/) = $arr;
                        }try {
                            if (empty($e)) {
                                self::getTcpPeer($addr)->send(['spawn_resolve',$pid, $result??null ]);
                            } else {
                                self::getTcpPeer($addr)->send(['spawn_reject',$pid, $e->getMessage()."\n".$e->getTraceAsString() ]);
                            }
                        } catch (\Throwable $e) {
                        }
                    }
                    break;
                default:
                    throw new \Exception(print_r($data, true), 1);
            }
        };
        if ($opt['proc_ttl']>1) {
            self::$lasttime=time();
            $event->add(rand(6,12), EventInterface::EV_TIMER, function () use ($opt) {
                if (self::$lasttime+$opt['proc_ttl']<time()) {
                    //echo "WORKER DIE\n";
                    exit;
                }
            });
        }
        $server->send(['worker']);
        $event->add(1, EventInterface::EV_TIMER, function () use ($server) {
            $server->send(['ping', array_keys(self::$tcp_peers) ]);
        });
        //echo "WORKER START\n";
        $event->loop();
        //echo "WORKER LOOPEND\n";
    }
}
